package com.diana.config;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;

import java.util.concurrent.*;

@Configuration
public class RocketMqProducerConfig {

    private static final Logger log = LoggerFactory.getLogger(RocketMqProducerConfig.class);


    /**
     * 普通消息生产者
     * @return
     * @throws MQClientException
     */
    /*@Bean
    public DefaultMQProducer defaultProducer() throws MQClientException {
        DefaultMQProducer producer = new DefaultMQProducer("dianaboot_group");
        producer.setNamesrvAddr("127.0.0.1:9876");
        producer.setVipChannelEnabled(false);
        producer.setRetryTimesWhenSendAsyncFailed(2);
        producer.start();
        log.info("rocketMq producer start-------------");
        return producer;
    }*/

    @Autowired
    private TransactionListenerImpl listener;

    /**
     * 事务消息生产者
     * @return
     * @throws MQClientException
     */
    @Bean
    @Primary
    public TransactionMQProducer transactionMQProducer() throws MQClientException {
        TransactionMQProducer producer = new TransactionMQProducer("diana_transation_mq_group");
        ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<>(100), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable runnable) {
                Thread thread = new Thread(runnable);
                thread.setName("rocketMQ_check_thread");
                return thread;
            }
        });
        producer.setNamesrvAddr("127.0.0.1:9876");
        producer.setExecutorService(executor);
        producer.setTransactionListener(listener);
        producer.start();
        log.info("rocketMq transactionProducer start-------------");
        return  producer;
    }
}
